package com.bmslz.monitor.base.grpc;

import com.bmslz.monitor.base.CommonConstant;
import com.bmslz.monitor.base.entity.MonitorModel;
import com.bmslz.monitor.base.entity.Project;
import com.bmslz.monitor.base.entity.ProjectData;
import com.bmslz.monitor.base.entity.User;
import com.bmslz.monitor.helper.LoginHelper;
import com.bmslz.monitor.service.IMonitorModelService;
import com.bmslz.monitor.service.IProjectDataService;
import com.bmslz.monitor.service.IProjectService;
import com.bmslz.monitor.service.IUserService;
import com.example.monitor.grpc.MetricsRequest;
import com.example.monitor.grpc.MetricsResponse;
import com.example.monitor.grpc.MonitorServiceGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

@Slf4j
@GrpcService
public class MonitorGrpcService extends MonitorServiceGrpc.MonitorServiceImplBase {

    public static final Map<Long, AtomicLong> qpsMap = new ConcurrentHashMap<>();

    @Scheduled(cron = "0 * * * * *")
    public void expiredClear() {
        long expiredTime = (System.currentTimeMillis() / 1000) - 60*5; // 保留最近5分钟的数据
        qpsMap.keySet().removeIf(time -> time < expiredTime);
    }

    @Autowired
    private IProjectDataService projectDataService;
    @Autowired
    private IProjectService projectService;
    @Autowired
    private IMonitorModelService monitorModelService;
    @Autowired
    private IUserService userService;

    /**
     * 这里是grpc通信消息的接收口
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void sendMetrics(MetricsRequest request, StreamObserver<MetricsResponse> responseObserver) {
        long currentSecond = System.currentTimeMillis() / 1000; // 获取当前秒级时间戳

        // 原子性更新当前秒的计数器
        qpsMap.computeIfAbsent(currentSecond, k -> new AtomicLong(0)).incrementAndGet();

        String projectNo = request.getProjectNo();//探针传来的项目编号
        String projectName = request.getProjectName();//探针传来的项目名称
        String modelNo = request.getModelNo();//探针传来的模型编号
        String projectRate = request.getProjectRate();//探针传来的上报频率

        //不做用户隔离
        LoginHelper.closeUserIsolation();
        //查找对应的监控模型
        MonitorModel monitorModel = monitorModelService.queryByNo(modelNo);
        if (monitorModel == null) {
            log.warn("监控模型编号不存在-{}", modelNo);
            responseObserver.onCompleted();
            return;
        }
        //获取当前用户
        String userId = monitorModel.getUserId();
        User user = userService.getById(userId);

        //从这里开始需要用户数据隔离 所以把上下文改为当前模型拥有者的id
        LoginHelper.openUserIsolation(user);
        //查询对应项目编号的项目 没有就新增 有就做一些update操作
        Project project = projectService.findByNo(projectNo);
        if (project == null) {
            project = new Project();
            project.setProjectNo(projectNo);//项目编号
            project.setProjectName(projectName);//项目名称
            project.setCreateTime(new Date());
            project.setUserId(monitorModel.getUserId());//用户id
            projectService.save(project);
        }
        Project updateProject = new Project();
        updateProject.setId(project.getId());
        updateProject.setHeartTime(new Date());//心跳时间
        updateProject.setMonitorModelId(monitorModel.getId());//监控模型id
        updateProject.setMonitorModelNo(modelNo);//监控模型编号
        updateProject.setProjectName(projectName);//项目名称
        updateProject.setProjectRate(Integer.valueOf(projectRate));//上报频率
        updateProject.setOnline(CommonConstant.ONLINE);//改为在线状态
        updateProject.setUpdateTime(new Date());
        projectService.updateById(updateProject);

        // 处理测点数据
        log.info("Received metrics from projectNo: " + projectNo);
        List<ProjectData> projectDataList = new ArrayList<>();
        Project finalProject = project;
        projectDataList = request.getMetricsList().stream().map(metric -> {
            log.info("Metric: {} = {} at {}",
                    metric.getName(), metric.getValue(), metric.getTimestamp());
            ProjectData data = new ProjectData();
            data.setUserId(monitorModel.getUserId());
            data.setAttKey(metric.getName());//测点key
            data.setAttName(monitorModelService.keyConvertName(monitorModel, metric.getName()));//测点名称
            data.setAttValue(metric.getValue());//测点数据
            data.setProjectId(finalProject.getId());//项目id
            data.setProjectNo(projectNo);//项目编号
            data.setProjectName(projectName);//项目名称
            data.setHeartTime(new Date(metric.getTimestamp()));//上报时间
            data.setCreateTime(new Date());
            return data;
        }).collect(Collectors.toList());
        //入库和告警
        projectDataService.handle(projectDataList, project);
        // 构建响应
        MetricsResponse response = MetricsResponse.newBuilder()
                .setSuccess(true)
                .setMessage("Metrics received successfully")
                .build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }
}